package nc

import (
	"context"
	"errors"
	"fmt"
	"strings"

	"github.com/project-flogo/core/data/metadata"
	"github.com/project-flogo/core/support/log"
	"github.com/project-flogo/core/trigger"

	"gitee.com/shipnet/flogo-common/netclient"
)

var triggerMd = trigger.NewMetadata(&Settings{}, &HandlerSettings{}, &Output{}, &Reply{})

func init() {
	_ = trigger.Register(&Trigger{}, &Factory{})
}

// Factory is a kafka trigger factory
type Factory struct {
}

// Metadata implements trigger.Factory.Metadata
func (*Factory) Metadata() *trigger.Metadata {
	return triggerMd
}

// New implements trigger.Factory.New
func (*Factory) New(config *trigger.Config) (trigger.Trigger, error) {
	s := &Settings{}
	err := metadata.MapToStruct(config.Settings, s, true)
	if err != nil {
		return nil, err
	}

	stx, err := parseDelimiter(s.STX)
	if err != nil {
		return nil, fmt.Errorf("failed parse stx: %w", err)
	}
	etx, err := parseDelimiter(s.ETX)
	if err != nil {
		return nil, fmt.Errorf("failed parse etx: %w", err)
	}
	max := s.Max
	framer, err := NewFramer(FramerConfig{
		STX:      stx,
		ETX:      etx,
		MaxBytes: max,
	})
	if err != nil {
		return nil, fmt.Errorf("failed create framer: %w", err)
	}

	host := s.Host
	port := s.Port
	if host == "" {
		return nil, errors.New("Valid host must be set")
	}
	if port == "" {
		return nil, errors.New("Valid port must be set")
	}

	t := Trigger{
		settings: s,
		framer:   framer,
	}
	return &t, nil
}

func parseDelimiter(s string) ([]byte, error) {
	dem := []byte{}
	bytes := []byte(s)
	for i := 0; i < len(bytes); i++ {
		b := bytes[i]
		if b == '\\' {
			i = i + 1
			if i >= len(bytes) {
				return []byte{}, fmt.Errorf("invalid delimiter bytes: %q, pos = %d", s, i-1)
			}
			b1 := bytes[i]
			switch b1 {
			case 'a':
				{
					dem = append(dem, '\a')
					break
				}
			case 'b':
				{
					dem = append(dem, '\b')
					break
				}
			case '\\':
				{
					dem = append(dem, '\\')
					break
				}
			case 't':
				{
					dem = append(dem, '\t')
					break
				}
			case 'f':
				{
					dem = append(dem, '\f')
					break
				}
			case 'r':
				{
					dem = append(dem, '\r')
					break
				}
			case 'v':
				{
					dem = append(dem, '\v')
					break
				}
			case '\'':
				{
					dem = append(dem, '\'')
					break
				}
			case '"':
				{
					dem = append(dem, '"')
					break
				}
			case 'n':
				{
					dem = append(dem, '\n')
					break
				}
			case 'x':
				{
					// read more 2 hex digit
					if i >= len(bytes)-2 {
						return []byte{}, fmt.Errorf("invalid delimiter bytes: %q pos = %d", s, i)
					}
					x1 := bytes[i+1]
					x2 := bytes[i+2]
					b, err := toByteHex(x1, x2)
					if err != nil {
						return []byte{}, fmt.Errorf("invalid dellimiter bytes: %q, pos = %d err = %w", s, i-2, err)
					}
					dem = append(dem, b)
					i = i + 2
					break
				}
			case 'X':
				{
					// read more 2 hex digit
					if i >= len(bytes)-2 {
						return []byte{}, fmt.Errorf("invalid delimiter bytes: %q pos = %d", s, i)
					}
					x1 := bytes[i+1]
					x2 := bytes[i+2]
					b, err := toByteHex(x1, x2)
					if err != nil {
						return []byte{}, fmt.Errorf("invalid delimiter bytes: %q, pos = %d err = %w", s, i-2, err)
					}
					dem = append(dem, b)
					i = i + 2
					break
				}
			default:
				{
					return []byte{}, fmt.Errorf("invalid delimiter byte: %q, pos = %d", s, i)
				}
			}

		} else {
			dem = append(dem, b)
		}
	}
	return dem, nil
}

func toByteNum(x byte) (byte, error) {
	if x >= '0' && x <= '9' {
		return x - '0', nil
	} else if x >= 'a' && x <= 'f' {
		return x - 'a' + 10, nil
	} else if x >= 'A' && x <= 'F' {
		return x - 'A' + 10, nil
	}
	return 0, fmt.Errorf("invalid hex digit: %c", x)
}

func toByteHex(x1 byte, x2 byte) (byte, error) {
	h1, err := toByteNum(x1)
	if err != nil {
		return 0, err
	}
	h2, err := toByteNum(x2)
	if err != nil {
		return 0, err
	}
	return (h1 << 4) + h2, nil
}

type clientHandler struct {
	handler  trigger.Handler
	settings *HandlerSettings
}

var matchEvents = map[string][]netclient.EventType{
	OnAny: []netclient.EventType{
		netclient.EventClose,
		netclient.EventConnected,
		netclient.EventData,
		netclient.EventDisconnect,
		netclient.EventError,
		netclient.EventPeerClose,
		netclient.EventReadTimeout,
		netclient.EventReconnect,
	},
	OnData:        []netclient.EventType{netclient.EventData},
	OnReconnect:   []netclient.EventType{netclient.EventReconnect},
	OnConnected:   []netclient.EventType{netclient.EventConnected},
	OnError:       []netclient.EventType{netclient.EventError},
	OnDisconnect:  []netclient.EventType{netclient.EventDisconnect},
	OnClose:       []netclient.EventType{netclient.EventClose},
	OnReadTimeout: []netclient.EventType{netclient.EventReadTimeout},
	OnPeerClose:   []netclient.EventType{netclient.EventPeerClose},
}

func (c *clientHandler) matchEventType(t netclient.EventType) bool {
	if es, ok := matchEvents[c.settings.Type]; ok {
		return contains(es, t)
	}
	return false
}

func contains(es []netclient.EventType, target netclient.EventType) bool {
	for _, t := range es {
		if t == target {
			return true
		}
	}
	return false
}

// Trigger is a kafka trigger
type Trigger struct {
	settings    *Settings
	handlers    []*clientHandler
	logger      log.Logger
	client      netclient.NetClient
	framer      Framer
	stop        chan int
	currentData []byte
}

// Initialize initializes the trigger
func (t *Trigger) Initialize(ctx trigger.InitContext) error {
	t.handlers = make([]*clientHandler, len(ctx.GetHandlers()))
	for idx, h := range ctx.GetHandlers() {
		s := HandlerSettings{}
		err := metadata.MapToStruct(h.Settings(), &s, true)
		if err != nil {
			return err
		}
		s.Type = strings.ToUpper(strings.TrimSpace(s.Type))
		ch := clientHandler{
			handler:  h,
			settings: &s,
		}
		t.handlers[idx] = &ch
	}
	t.logger = ctx.Logger()

	// t.Conn = conn
	t.stop = make(chan int, 1)

	return nil
}

// Start starts the kafka trigger
func (t *Trigger) Start() error {
	go t.connect()
	t.logger.Infof("connect to listener on Port - %s, Network - %s", t.settings.Port, t.settings.Network)
	return nil
}

func (t *Trigger) connect() {
	host := t.settings.Host
	port := t.settings.Port
	addr := fmt.Sprintf("%s://%s:%s", t.settings.Network, host, port)
	s := netclient.Settings{
		Addr:              addr,
		MaxSize:           t.settings.Max,
		Timeout:           t.settings.Timeout,
		ConnectionDelay:   "5s",
		ReconnectInterval: "5s",
		Logger:            t.logger,
	}
	var err error
	t.client, err = netclient.New(s)
	if err != nil {
		t.logger.Errorf("failed create net client: %v", err)
		return
	}
	r, err := t.client.Receive()
	if err != nil {
		t.logger.Errorf("failed starting receiving data: %v", err)
		return
	}
	ch, err := r.Events()
	if err != nil {
		t.logger.Errorf("failed get channel: %v", err)
		return
	}
	for event := range ch {
		switch event.Type {
		case netclient.EventData:
			{
				rs := t.getHandlers(event.Type)
				t.handleEventData(rs, event)
				break
			}
		case netclient.EventReconnect:
			{
				rs := t.getHandlers(event.Type)
				t.handleEventReconnect(rs, event)
				break
			}
		case netclient.EventConnected:
			{
				rs := t.getHandlers(event.Type)
				t.handleEventConnected(rs, event)
				break
			}
		case netclient.EventError:
			{
				rs := t.getHandlers(event.Type)
				t.handleEventError(rs, event)
				break
			}
		case netclient.EventDisconnect:
			{
				rs := t.getHandlers(event.Type)
				t.handleEventDisconnect(rs, event)
				break
			}
		case netclient.EventClose:
			{
				rs := t.getHandlers(event.Type)
				t.handleEventClose(rs, event)
				break
			}
		case netclient.EventReadTimeout:
			{
				rs := t.getHandlers(event.Type)
				t.handleEventReadTimeout(rs, event)
				break
			}
		case netclient.EventPeerClose:
			{
				rs := t.getHandlers(event.Type)
				t.handleEventPeerClose(rs, event)
				break
			}
		default:
			{
				rs := []*clientHandler{}
				for _, h := range t.handlers {
					if h.settings.Type == OnAny {
						rs = append(rs, h)
					}
				}
				t.handleUnknownEvent(rs, event)
				break
			}
		}

	}
}

func (t *Trigger) getHandlers(eventType netclient.EventType) []*clientHandler {
	receivers := []*clientHandler{}
	for _, h := range t.handlers {
		if h.matchEventType(eventType) {
			receivers = append(receivers, h)
		}
	}
	return receivers
}

func (t *Trigger) handleEventData(handlers []*clientHandler, event netclient.Event) {
	allBytes := append(t.currentData, event.Data...)
	result := t.framer.Frame(allBytes)
	t.currentData = result.Remains()
	for _, frame := range result.Frames() {
		t.handleFrame(handlers, frame, event)
	}
}

func (t *Trigger) handleFrame(handlers []*clientHandler, data []byte, event netclient.Event) {
	output := Output{
		Type:                OnData,
		Data:                data,
		LocalNetwork:        event.LocalNetwork,
		LocalAddr:           event.LocalAddr,
		RemoteNetwork:       event.RemoteNetwork,
		RemoteAddr:          event.RemoteAddr,
		TotalConnectedCount: event.TotalConnectedCount,
		RecentRetryCount:    event.RecentRetryCount,
		FirstConnectedTime:  event.FirstConnectedTime,
		LastConnectedTime:   event.LastConnectedTime,
	}

	for _, h := range handlers {
		m, err := h.handler.Handle(context.Background(), &output)
		if err != nil {
			t.logger.Errorf("failed running handler name = %s err = %v", h.handler.Name(), err)
		} else {
			reply := &Reply{}
			err := reply.FromMap(m)
			if err != nil {
				t.logger.Errorf("invalid reply data name = %s err = %v", h.handler.Name(), err)
			} else {
				err := t.client.Send(reply.Reply)
				if err != nil {
					t.logger.Errorf("failed send reply name = %s err = %v", h.handler.Name(), err)
				}
			}
		}

	}
}

func (t *Trigger) handleEventReconnect(handlers []*clientHandler, event netclient.Event) {
	output := Output{
		Type:                OnReconnect,
		Data:                []byte{},
		LocalNetwork:        event.LocalNetwork,
		LocalAddr:           event.LocalAddr,
		RemoteNetwork:       event.RemoteNetwork,
		RemoteAddr:          event.RemoteAddr,
		TotalConnectedCount: event.TotalConnectedCount,
		RecentRetryCount:    event.RecentRetryCount,
		FirstConnectedTime:  event.FirstConnectedTime,
		LastConnectedTime:   event.LastConnectedTime,
	}
	for _, h := range handlers {
		m, err := h.handler.Handle(context.Background(), &output)
		if err != nil {
			t.logger.Errorf("failed running handler name = %s err = %v", h.handler.Name(), err)
		} else {
			t.logger.Debugf("succesfully running handler name = %s err = %v result = %v", h.handler.Name(), err, m)
		}
	}
}

func (t *Trigger) handleEventConnected(handlers []*clientHandler, event netclient.Event) {
	output := Output{
		Type:                OnReconnect,
		Data:                []byte{},
		LocalNetwork:        event.LocalNetwork,
		LocalAddr:           event.LocalAddr,
		RemoteNetwork:       event.RemoteNetwork,
		RemoteAddr:          event.RemoteAddr,
		TotalConnectedCount: event.TotalConnectedCount,
		RecentRetryCount:    event.RecentRetryCount,
		FirstConnectedTime:  event.FirstConnectedTime,
		LastConnectedTime:   event.LastConnectedTime,
	}
	for _, h := range handlers {
		m, err := h.handler.Handle(context.Background(), &output)
		if err != nil {
			t.logger.Errorf("failed running handler name = %s err = %v", h.handler.Name(), err)
		} else {
			t.logger.Debugf("succesfully running handler name = %s err = %v result = %v", h.handler.Name(), err, m)
		}
	}
}

func (t *Trigger) handleEventError(handlers []*clientHandler, event netclient.Event) {
	output := Output{
		Type:                OnError,
		Data:                []byte{},
		LocalNetwork:        event.LocalNetwork,
		LocalAddr:           event.LocalAddr,
		RemoteNetwork:       event.RemoteNetwork,
		RemoteAddr:          event.RemoteAddr,
		TotalConnectedCount: event.TotalConnectedCount,
		RecentRetryCount:    event.RecentRetryCount,
		FirstConnectedTime:  event.FirstConnectedTime,
		LastConnectedTime:   event.LastConnectedTime,
	}
	for _, h := range handlers {
		m, err := h.handler.Handle(context.Background(), &output)
		if err != nil {
			t.logger.Errorf("failed running handler name = %s err = %v", h.handler.Name(), err)
		} else {
			t.logger.Debugf("succesfully running handler name = %s err = %v result = %v", h.handler.Name(), err, m)
		}
	}
}

func (t *Trigger) handleEventReadTimeout(handlers []*clientHandler, event netclient.Event) {
	output := Output{
		Type:                OnReadTimeout,
		Data:                []byte{},
		LocalNetwork:        event.LocalNetwork,
		LocalAddr:           event.LocalAddr,
		RemoteNetwork:       event.RemoteNetwork,
		RemoteAddr:          event.RemoteAddr,
		TotalConnectedCount: event.TotalConnectedCount,
		RecentRetryCount:    event.RecentRetryCount,
		FirstConnectedTime:  event.FirstConnectedTime,
		LastConnectedTime:   event.LastConnectedTime,
	}
	for _, h := range handlers {
		m, err := h.handler.Handle(context.Background(), &output)
		if err != nil {
			t.logger.Errorf("failed running handler name = %s err = %v", h.handler.Name(), err)
		} else {
			t.logger.Debugf("succesfully running handler name = %s err = %v result = %v", h.handler.Name(), err, m)
		}
	}
}

func (t *Trigger) handleEventPeerClose(handlers []*clientHandler, event netclient.Event) {
	output := Output{
		Type:                OnPeerClose,
		Data:                []byte{},
		LocalNetwork:        event.LocalNetwork,
		LocalAddr:           event.LocalAddr,
		RemoteNetwork:       event.RemoteNetwork,
		RemoteAddr:          event.RemoteAddr,
		TotalConnectedCount: event.TotalConnectedCount,
		RecentRetryCount:    event.RecentRetryCount,
		FirstConnectedTime:  event.FirstConnectedTime,
		LastConnectedTime:   event.LastConnectedTime,
	}
	for _, h := range handlers {
		m, err := h.handler.Handle(context.Background(), &output)
		if err != nil {
			t.logger.Errorf("failed running handler name = %s err = %v", h.handler.Name(), err)
		} else {
			t.logger.Debugf("succesfully running handler name = %s err = %v result = %v", h.handler.Name(), err, m)
		}
	}
}

func (t *Trigger) handleEventClose(handlers []*clientHandler, event netclient.Event) {
	output := Output{
		Type:                OnClose,
		Data:                []byte{},
		LocalNetwork:        event.LocalNetwork,
		LocalAddr:           event.LocalAddr,
		RemoteNetwork:       event.RemoteNetwork,
		RemoteAddr:          event.RemoteAddr,
		TotalConnectedCount: event.TotalConnectedCount,
		RecentRetryCount:    event.RecentRetryCount,
		FirstConnectedTime:  event.FirstConnectedTime,
		LastConnectedTime:   event.LastConnectedTime,
	}
	for _, h := range handlers {
		m, err := h.handler.Handle(context.Background(), &output)
		if err != nil {
			t.logger.Errorf("failed running handler name = %s err = %v", h.handler.Name(), err)
		} else {
			t.logger.Debugf("succesfully running handler name = %s err = %v result = %v", h.handler.Name(), err, m)
		}
	}
}

func (t *Trigger) handleEventDisconnect(handlers []*clientHandler, event netclient.Event) {
	output := Output{
		Type:                OnDisconnect,
		Data:                []byte{},
		LocalNetwork:        event.LocalNetwork,
		LocalAddr:           event.LocalAddr,
		RemoteNetwork:       event.RemoteNetwork,
		RemoteAddr:          event.RemoteAddr,
		TotalConnectedCount: event.TotalConnectedCount,
		RecentRetryCount:    event.RecentRetryCount,
		FirstConnectedTime:  event.FirstConnectedTime,
		LastConnectedTime:   event.LastConnectedTime,
	}
	for _, h := range handlers {
		m, err := h.handler.Handle(context.Background(), &output)
		if err != nil {
			t.logger.Errorf("failed running handler name = %s err = %v", h.handler.Name(), err)
		} else {
			t.logger.Debugf("succesfully running handler name = %s err = %v result = %v", h.handler.Name(), err, m)
		}
	}
}

func (t *Trigger) handleUnknownEvent(handlers []*clientHandler, event netclient.Event) {
	output := Output{
		Type:                OnAny,
		Data:                []byte{},
		LocalNetwork:        event.LocalNetwork,
		LocalAddr:           event.LocalAddr,
		RemoteNetwork:       event.RemoteNetwork,
		RemoteAddr:          event.RemoteAddr,
		TotalConnectedCount: event.TotalConnectedCount,
		RecentRetryCount:    event.RecentRetryCount,
		FirstConnectedTime:  event.FirstConnectedTime,
		LastConnectedTime:   event.LastConnectedTime,
	}
	for _, h := range handlers {
		m, err := h.handler.Handle(context.Background(), &output)
		if err != nil {
			t.logger.Errorf("failed running handler name = %s err = %v", h.handler.Name(), err)
		} else {
			t.logger.Debugf("succesfully running handler name = %s err = %v result = %v", h.handler.Name(), err, m)
		}
	}
}

// Stop implements ext.Trigger.Stop
func (t *Trigger) Stop() error {
	t.client.Close()

	t.logger.Info("Stopped listener")

	return nil
}
